package com.infra.micro.id.domain.core.segment;

import com.infra.micro.id.domain.entity.IdSegmentPO;
import com.infra.micro.id.domain.core.AbstractIdGenerator;
import com.infra.micro.id.domain.repository.IdSegmentRepository;
import org.redisson.api.RLock;
import org.redisson.api.RQueue;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.LongStream;

/**
 * 号段模式id生成器
 *
 * @author PD
 **/
@Component
public class IdSegmentGenerator extends AbstractIdGenerator {

    private static final String SEGMENT_BUFFER_CACHE_KEY = "ID-SERVICE:SEGMENT-BUFFER";
    private static final int BUFFER_MIN_LENGTH = 500;
    protected final IdSegmentRepository idSegmentRepository;

    public IdSegmentGenerator(IdSegmentRepository idSegmentRepository) {
        this.idSegmentRepository = idSegmentRepository;
    }

    @Override
    public String name() {
        return "segment";
    }

    @Override
    public Long getId(Integer appId) {
        final String Key = String.format("%s-%s", SEGMENT_BUFFER_CACHE_KEY, appId.toString());

        RLock lock = redissonClient.getLock(String.format("%s:LOCK", Key));
        lock.lock(2, TimeUnit.SECONDS);

        try {
            // 队列
            RQueue<Object> queue = redissonClient.getQueue(Key);
            if (queue.size() == 0) {
                queue.addAll(buildSegmentList(appId, queue, idSegmentRepository.updateMaxIdAndGet(appId)));
            }
            // 取出数值
            return Long.class.cast(queue.poll());
        } finally {
            lock.unlock();
        }

    }

    @Override
    public Set<Long> getIds(Integer appId, int step) {
        final String Key = String.format("%s-%s", SEGMENT_BUFFER_CACHE_KEY, appId.toString());

        RLock lock = redissonClient.getLock(String.format("%s:LOCK", Key));
        lock.lock(10, TimeUnit.SECONDS);

        // 队列
        RQueue<Object> queue = redissonClient.getQueue(Key);
        int queueSize = queue.size();

        try {
            //不从缓存取，直接生成当前批次
            if (queueSize > step) {
                LOGGER.info("从appId:{}号段缓存队中取{}个ID号", appId, step);
                return castSet(queue.poll(step), Long.class);
            }

            Set<Long> result = new HashSet<>(step);
            IdSegmentPO idSegmentPO = idSegmentRepository.updateMaxIdByStepAndGet(appId, step);
            LongStream.range(idSegmentPO.getMaxId().longValue() - step, idSegmentPO.getMaxId().longValue())
                    .mapToObj(i -> i + 1)
                    .parallel().forEach(result::add);

            if (queue.size() < BUFFER_MIN_LENGTH) {
                queue.addAll(buildSegmentList(appId, queue, idSegmentRepository.updateMaxIdAndGet(appId)));
            }

            return result;
        } finally {
            lock.unlock();
        }
    }

    private Set<Long> buildSegmentList(Integer appId, RQueue<Object> queue, IdSegmentPO idSegmentPO) {
        long startId = idSegmentPO.getMaxId().longValue() - idSegmentPO.getStep().longValue();
        LOGGER.info("appId:{} 缓存号段个数:{},号段不足已支持交付，预批号段[{},{}]已生成加入到缓存队列", appId, queue.size(), startId, idSegmentPO.getMaxId());

        return LongStream.range(startId, idSegmentPO.getMaxId().longValue())
                .parallel()
                .mapToObj(i -> i + 1)
                .collect(Collectors.toSet());
    }

}
